package transcode

import (
	"context"
	"encoding/json"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mmetrics"

	"code.byted.org/bytertc/whiteboard_server/biz/api"
	"code.byted.org/bytertc/whiteboard_server/biz/config"
	"code.byted.org/bytertc/whiteboard_server/biz/dal"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/gateway"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mq"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/rabbit_mq"
	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/callback"
	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/task"
	"code.byted.org/gopkg/env"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/gopkg/metrics"
	"code.byted.org/videoarch/universal_utils_go/event_collector/tracer"
	"code.byted.org/videoarch/universal_utils_go/event_collector/util"
	"github.com/google/uuid"
	"github.com/pkg/errors"
	"github.com/spf13/cast"
)

type Manager interface {
	CreateTask(ctx context.Context, request *api.CreateTranscodeTaskRequest) (string, error)
	QueryTask(ctx context.Context, request *api.QueryTranscodeTaskRequest) (int, int, int, string, error)
	GetResult(ctx context.Context, request *api.GetTranscodeTaskRequest) (int, *callback.Result, error)
	HandleExecutorCallback(ctx context.Context, event *callback.Event) error
	HandleDynamicCallback(ctx context.Context, dynamicCallback *callback.DynamicCallback) error
	HandlerException(ctx context.Context) error
	HandleWpsCallbackFileDetail(ctx context.Context, request *api.WpsCallbackFileDetailRequest) (*api.WpsCallbackFileDetailResponse, error)
	HandleWpsCallbackDownload(ctx context.Context, request *api.WpsCallbackDownloadRequest) (*api.WpsCallbackDownloadResponse, error)
	HandleWpsCallbackPermission(ctx context.Context, request *api.WpsCallbackPermissionRequest) (*api.WpsCallbackPermissionResponse, error)
	HandleWpsCallbackNotify(ctx context.Context, request *api.WpsCallbackNotifyRequest) error
	GetResource(ctx context.Context, request *api.TranscodeGetResourceRequest) (*api.TranscodeGetResourceResponse, error)
}

func NewManager(repo dal.Repo, config config.DynamicConfig, rabbitMq *rabbit_mq.Client, rtcAPI gateway.RtcAPI, metricsClient *metrics.MetricsClientV2, topic string, sender mq.KafkaSender) Manager {
	return &manager{
		repo:          repo,
		config:        config,
		rabbitMq:      rabbitMq,
		rtcAPI:        rtcAPI,
		metricsClient: metricsClient,
		BillTopic:     topic,
		BillSender:    sender,
	}
}

type manager struct {
	repo          dal.Repo
	config        config.DynamicConfig
	rabbitMq      *rabbit_mq.Client
	rtcAPI        gateway.RtcAPI
	metricsClient *metrics.MetricsClientV2
	BillTopic     string
	BillSender    mq.KafkaSender
}

func (m *manager) CreateTask(ctx context.Context, request *api.CreateTranscodeTaskRequest) (string, error) {
	//todo bytedtrace.span
	t := buildTask(request)
	log.V1.CtxInfo(ctx, "create task:%#v", t)
	if t.StorageConfig.Type == task.StoreInTos {
		if t.StorageConfig.TosConfig.AccountId == "" || t.StorageConfig.TosConfig.Bucket == "" {
			consoleTosConfig, err := m.getConsoleTosConfig(ctx, t.AppId)
			if err != nil {
				return "", err
			}
			t.StorageConfig.TosConfig = *consoleTosConfig
		}
	}
	if t.TranscodeMode == task.Dynamic {
		err := t.DynamicCheckAndUpdate()
		if err != nil {
			log.V1.CtxError(ctx, "check param failed,task:%#v,err:%s", t, err)
			return "", err
		}
		err = m.repo.CreateTranscodeTask(ctx, t.ToPersist())
		if err != nil {
			log.V1.CtxError(ctx, "db: create task failed,task:%v,error:%s", err, t)
			return "", biz_error.TranscodeError.Wrap(err)
		}

		et, err := t.ToExecutorTask()
		if err != nil {
			log.V1.CtxError(ctx, "convert to executor task failed,task:%#v,err:%s", t, err)
			return "", err
		}

		token, err := m.rtcAPI.GenByteToken(ctx, t.AppId, "wps", "wps")
		if err != nil {
			return "", biz_error.TranscodeError.Wrap(err)
		}

		sttRequest := &startTranscodeTaskRequest{
			AppID:  t.AppId,
			TaskID: t.TaskId,
			FileID: t.FileID,
			Token:  token,
			Callback: callbackInfo{
				URL: config.GetConfig().DynamicCallback,
			},
		}
		if t.TranscodeConfig.Thumbnail {
			sttRequest.Thumbnail = &thumbnail{
				ForceUseResolution:    et.TranscodeConfig.ForceUseResolution,
				ThumbnailFormat:       int(et.TranscodeConfig.OutputFormat),
				ThumbnailHeight:       et.TranscodeConfig.ThumbnailHeight,
				ThumbnailWidth:        et.TranscodeConfig.ThumbnailWidth,
				PreSignDuration:       t.PreSignDuration,
				ExecutorStorageConfig: et.ExecutorStorageConfig,
			}
		}
		sttResp, err := startDynamicTranscodeTask(ctx, sttRequest)
		if err != nil {
			log.V1.CtxError(ctx, "start transcode error:%+v,task:%+v", err, t)
			return "", errors.WithStack(err)
		}
		if !sttResp.Success {
			tags := []metrics.T{
				{Name: "app_id", Value: t.AppId},
			}
			err = m.metricsClient.EmitCounter(mmetrics.MetricsDynamicStartError, 1, tags...)
			if err != nil {
				log.V1.CtxError(ctx, "Monitor EmitCounter failed,error:%v", err)
			}
			return "", errors.New("not success")
		}

		traceKey := &util.KeyWord{
			AppId:  t.AppId,
			UserId: t.UserId,
			TaskId: t.TaskId,
		}
		otherInfo := map[string]string{
			"psm":            env.PSM(),
			"transcode_mode": cast.ToString(int(t.TranscodeMode)),
			"input_format":   t.TranscodeConfig.GetInputFormat(),
			"output_format":  t.TranscodeConfig.GetOutputFormat(),
			"priority":       cast.ToString(int(t.Priority)),
		}
		tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_start", "", traceKey, otherInfo)

		return t.TaskId, nil
	}
	err := t.CheckAndUpdate()
	if err != nil {
		log.V1.CtxError(ctx, "check param failed,task:%#v,err:%s", t, err)
		return "", err
	}

	et, err := t.ToExecutorTask()
	if err != nil {
		log.V1.CtxError(ctx, "convert to executor task failed,task:%#v,err:%s", t, err)
		return "", err
	}
	msg, err := json.Marshal(et)
	if err != nil {
		log.V1.CtxError(ctx, "json error,err:%s", err)
		return "", biz_error.TranscodeError.Wrap(err)
	}

	err = m.repo.CreateTranscodeTask(ctx, t.ToPersist())
	if err != nil {
		log.V1.CtxError(ctx, "db: create task failed,task:%v,error:%s", err, t)
		return "", biz_error.TranscodeError.Wrap(err)
	}

	switch t.Priority {
	//notice 切第三方sdk使用lambda后，实时任务可以直接启动一个lambda,非实时任务可存到kafka，在加一个调度服务，将这些任务调度到有限的lambda实例上
	case task.RealTime:
		err = m.rabbitMq.Publish(ctx, msg, rabbit_mq.HighPriority)
	default:
		err = m.rabbitMq.Publish(ctx, msg, rabbit_mq.LowPriority)
	}
	if err != nil {
		log.V1.CtxError(ctx, "rabbit_mq publish msg failed,error:%s,task:%+v", err, t)
		return "", biz_error.TranscodeError.Wrap(err)
	}

	traceKey := &util.KeyWord{
		AppId:  t.AppId,
		UserId: t.UserId,
		TaskId: t.TaskId,
	}
	otherInfo := map[string]string{
		"transcode_mode": cast.ToString(int(t.TranscodeMode)),
		"input_format":   t.TranscodeConfig.GetInputFormat(),
		"output_format":  t.TranscodeConfig.GetOutputFormat(),
		"priority":       cast.ToString(int(t.Priority)),
	}
	tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_start", "", traceKey, otherInfo)

	return t.TaskId, nil
}

func (m *manager) QueryTask(ctx context.Context, request *api.QueryTranscodeTaskRequest) (int, int, int, string, error) {
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  request.AppId,
		TaskId: request.TaskId,
	})
	if err != nil {
		return 0, 0, 0, "", err
	}
	if t.Status == int(task.Failed) {
		errMsg := ""
		switch t.ErrCode {
		case 400:
			errMsg = "param error"
		case 401:
			errMsg = "file size exceed limit"
		case 402:
			errMsg = "pages count exceed limit"
		case 420:
			errMsg = "download failed"
		case 421:
			errMsg = "download failed,server error"
		case 440:
			errMsg = "transcode failed"
		case 460:
			errMsg = "upload failed"
		case 500:
			errMsg = "internal error"
		}
		if t.TranscodeMode == int(task.Dynamic) {
			errMsg = t.ErrMsg
		}
		return t.Status, 0, t.ErrCode, errMsg, nil
	}

	progress := 0
	if t.TotalPages != 0 {
		progress = (t.CurPages * 100) / t.TotalPages
	}
	return t.Status, progress, 0, "", nil
}

func (m *manager) GetResult(ctx context.Context, request *api.GetTranscodeTaskRequest) (int, *callback.Result, error) {
	log.V1.CtxInfo(ctx, "get result,request:%#v", request)
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  request.AppId,
		TaskId: request.TaskId,
	})
	if err != nil {
		return 0, nil, biz_error.TranscodeError.Wrap(err)
	}
	if t.Status != int(task.Done) {
		return 0, nil, biz_error.TranscodeError.Wrap(errors.New("task is not done"))
	}

	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return 0, nil, biz_error.TranscodeError.Wrap(err)
	}

	rs, err := m.repo.GetTranscodeResult(ctx, &task.TaskUnique{
		AppId:  request.AppId,
		TaskId: request.TaskId,
	})
	if err != nil {
		return 0, nil, biz_error.TranscodeError.Wrap(err)
	}
	if t.TranscodeMode == int(task.Statistic) && len(rs) == 0 {
		return 0, nil, biz_error.TranscodeError.Wrap(errors.New("result is empty"))
	}

	res := &callback.Result{
		Width:         tt.TranscodeConfig.OutputWidth,
		Height:        tt.TranscodeConfig.OutputHeight,
		TranscodeMode: t.TranscodeMode,
	}
	if t.TranscodeMode == int(task.Dynamic) {
		res.FileID = t.FileID
		res.FileName = t.FileName
	} else {
		t.FileName = rs[0].FileName
	}
	for _, r := range rs {
		res.Images = append(res.Images, callback.Image{
			PageID:       r.PageID,
			Img:          r.ImgUrl,
			ThumbnailUrl: r.ThmUrl,
		})
	}

	return t.TranscodeMode, res, nil
}

func (m *manager) HandleExecutorCallback(ctx context.Context, event *callback.Event) error {
	log.V1.CtxInfo(ctx, "handle executor callback,evnet:%#v", event)
	var err error
	switch event.EventType {
	case callback.TaskStart:
		err = m.handleTaskStart(ctx, event.EventData)
	case callback.TaskProcess:
		err = m.handleTaskProcess(ctx, event.EventData)
	case callback.TaskFinish:
		err = m.handleTaskFinish(ctx, event.EventData)
	case callback.Stats:
		err = m.handleStats(ctx, event.EventData)
	}
	if err != nil {
		log.V1.CtxError(ctx, "handle callback failed,event:%#v,error:%s", event, err)
	}
	//回调处理失败，返回错误也没用
	return nil
}

func (m *manager) HandleDynamicCallback(ctx context.Context, dynamicCallback *callback.DynamicCallback) error {
	if dynamicCallback.AppID == "" || dynamicCallback.TaskID == "" {
		return nil
	}
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  dynamicCallback.AppID,
		TaskId: dynamicCallback.TaskID,
	})
	if err != nil {
		return err
	}
	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return err
	}
	result := "success"
	if dynamicCallback.Success {
		err = m.repo.UpdateTranscodeTask(ctx,
			&task.TaskUnique{
				AppId:  t.AppId,
				TaskId: t.TaskId},
			map[string]interface{}{
				"status":      task.Done,
				"total_pages": dynamicCallback.Result.FileInfo.PptPageCount,
				"cur_pages":   dynamicCallback.Result.FileInfo.PptPageCount,
				"ratio":       dynamicCallback.Result.FileInfo.PptRatio,
			})
		if err != nil {
			log.V1.CtxError(ctx, "UpdateTranscodeTask failed,task:%+v,err:%+v", t, err)
			return err
		}
		if len(dynamicCallback.Result.Thumbnails) > 0 {
			rs := make([]*ddl.TranscodeResult, 0)
			for _, r := range dynamicCallback.Result.Thumbnails {
				rs = append(rs, &ddl.TranscodeResult{
					AppId:    t.AppId,
					TaskId:   t.TaskId,
					FileName: t.FileName,
					PageID:   r.PageIndex,
					ThmUrl:   r.ThumbnailURL,
				})
			}
			err = m.repo.BatchCreateTranscodeResult(ctx, rs)
			if err != nil {
				log.V1.CtxError(ctx, "BatchCreateTranscodeResult failed,task:%+v,err:%+v", t, err)
				return errors.WithStack(err)
			}
		}
		err = m.sendToBill(ctx, t.AppId, t.TaskId, t.TranscodeMode, t.TotalPages)
		if err != nil {
			log.V1.CtxError(ctx, "send to bill failed,task:%+v,err:%+v", t, err)
		}
	} else {
		result = "fail"
		err = m.repo.UpdateTranscodeTask(ctx,
			&task.TaskUnique{
				AppId:  t.AppId,
				TaskId: t.TaskId},
			map[string]interface{}{
				"status":   task.Failed,
				"err_code": dynamicCallback.ErrCode,
				"err_msg":  dynamicCallback.ErrMsg,
			})
		if err != nil {
			log.V1.CtxError(ctx, "UpdateTranscodeTask failed,task:%+v,err:%+v", t, err)
			return err
		}
	}

	traceKey := &util.KeyWord{
		AppId:  t.AppId,
		UserId: t.UserId,
		TaskId: t.TaskId,
	}
	otherInfo := map[string]string{
		"transcode_mode": cast.ToString(int(tt.TranscodeMode)),
		"input_format":   "",
		"output_format":  "",
		"priority":       cast.ToString(int(tt.Priority)),
		"size":           cast.ToString(t.FileSize),
		"total_pages":    cast.ToString(t.TotalPages),
		"cur_pages":      cast.ToString(t.TotalPages),
		"result":         result,
		"fail_reason":    cast.ToString(dynamicCallback.ErrCode),
	}
	tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_finish", "", traceKey, otherInfo)
	return nil
}

func buildTask(request *api.CreateTranscodeTaskRequest) *task.Task {
	t := &task.Task{
		TaskUnique: task.TaskUnique{
			AppId:  request.AppId,
			TaskId: uuid.New().String(),
		},
		UserId:          request.Operator,
		Status:          task.Submited,
		Resource:        request.Resource,
		ResourceAttr:    request.ResourceAttr,
		TranscodeMode:   request.TranscodeMode,
		Priority:        request.Priority,
		TranscodeConfig: request.TranscodeConfig,
		StorageConfig:   request.StorageConfig,
		PreSignDuration: request.PreSignDuration,
	}
	return t
}

func (m *manager) getConsoleTosConfig(ctx context.Context, appID string) (*task.TosConfig, error) {
	sr := &api.SwitchStatusRequest{
		AppID:    appID,
		Category: dal.Category_Switch_Transcode,
	}
	switchInfos, err := m.repo.GetSwitchInfo(ctx, sr)
	if err != nil {
		return nil, err
	}
	if len(switchInfos) == 0 {
		return nil, biz_error.InvalidParam.Wrap(errors.Errorf("not find any console config"))
	}
	if switchInfos[0].Status == dal.Switch_OFF {
		return nil, biz_error.InvalidParam.Wrap(errors.Errorf("console swtich not open"))
	}
	cr := &api.GetConsoleInfoRequest{
		AppID:    appID,
		Category: dal.Category_File_Transcode,
	}

	configInfos, err := m.repo.GetConfig(ctx, cr, false)
	if err != nil {
		return nil, err
	}
	if len(configInfos) == 0 {
		return nil, biz_error.InvalidParam.Wrap(errors.Errorf("not find any tr transcode config"))
	}
	spaceType := configInfos[0].SpaceType
	space := configInfos[0].Space
	if space == "" {
		return nil, biz_error.InvalidParam.Wrap(errors.Errorf("config sapce is empty"))
	}
	accountId := configInfos[0].AccountId
	if accountId == "" {
		return nil, biz_error.InvalidParam.Wrap(errors.Errorf("config account is empty"))
	}

	if spaceType == dal.Space_Type_TOS {
		res := &task.TosConfig{
			AccountId: accountId,
			Bucket:    space,
			Region:    0,
		}
		return res, nil
	}
	return nil, biz_error.InvalidParam.Wrap(errors.Errorf("config not found"))
}

func (m *manager) GetResource(ctx context.Context, request *api.TranscodeGetResourceRequest) (*api.TranscodeGetResourceResponse, error) {
	if request.FileID == "" {
		return nil, errors.New("file_id is empty")
	}
	t, err := m.repo.GetTranscodeTaskByFileID(ctx, request.FileID)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	if t.Status != int(task.Done) {
		return nil, errors.New("task not success")
	}
	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	return &api.TranscodeGetResourceResponse{
		FileID:    request.FileID,
		Resource:  tt.Resource,
		PageCount: t.TotalPages,
		FileRatio: t.Ratio,
	}, nil
}

func (m *manager) HandlerException(ctx context.Context) error {
	return nil
}
